package com.gitee.wsl.flow.create

import com.gitee.wsl.flow.F
import com.gitee.wsl.flow.transform.map.FlatMapStrategy
import com.gitee.wsl.flow.transform.map.flatMap
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.conflate
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.isActive
import kotlin.time.ComparableTimeMark
import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds
import kotlin.time.ExperimentalTime
import kotlin.time.TimeSource

/**
 * Creates a [Flow] that will wait for a specified time, before emitting the [value].
 */
fun <T> F.timerDelay(value: T, timeMillis: Long): Flow<T> = flow {
    delay(timeMillis)
    emit(value)
}

/**
 * Creates a [Flow] that will wait for a given [duration], before emitting the [value].
 */
fun <T> F.timerDelay(value: T, duration: Duration): Flow<T> = flow {
    delay(duration)
    emit(value)
}

/**
 * Returns a [Flow] that emits a 0L after the [initialDelay] and ever-increasing numbers
 * after each [period] of time thereafter.
 *
 * @param initialDelay must be greater than or equal to [Duration.ZERO]
 * @param period must be greater than or equal to [Duration.ZERO]
 */
fun F.timeInterval(
    initialDelay: Duration,
    period: Duration,
): Flow<Long> {
    require(initialDelay >= Duration.ZERO) { "Expected non-negative delay, but has $initialDelay ms" }
    require(period >= Duration.ZERO) { "Expected non-negative period, but has $period ms" }

    return flow {
        delay(initialDelay)

        var count = 0L
        while (true) {
            emit(count++)
            delay(period)
        }
    }
}

/**
 * Returns a [Flow] that emits a 0L after the [initialDelayMillis] and ever-increasing numbers
 * after each [periodMillis] of time thereafter.
 *
 * @param initialDelayMillis must be non-negative
 * @param periodMillis must be non-negative
 */
fun F.timeInterval(
    initialDelayMillis: Long,
    periodMillis: Long,
): Flow<Long> {
    require(initialDelayMillis >= 0) { "Expected non-negative delay, but has $initialDelayMillis ms" }
    require(periodMillis >= 0) { "Expected non-negative periodMillis, but has $periodMillis ms" }

    return flow {
        delay(initialDelayMillis)

        var count = 0L
        while (true) {
            emit(count++)
            delay(periodMillis)
        }
    }
}

fun <T> interval(time: Long, initialValue: T, newValue: (T) -> T): Flow<T> {
    var value = initialValue
    return interval(time)
        .map {
            value.apply {
                value = newValue(this)
            }
        }
}

fun interval(time: Long, timeOut: Long = time) = flow {
    if (timeOut > 0) {
        delay(timeOut)
    }
    emit(Unit)
    while (true) {
        delay(time)
        emit(Unit)
    }
}

/**
 * Retrieves a [Flow] of [Unit] that emits after the provided [delay] and then finishes.
 *
 * @param [delay] The [Duration] to wait before emitting a [Unit] and then completing.
 *
 * @author chRyNaN
 */
@ExperimentalTime
@ExperimentalCoroutinesApi
fun F.timerFlow(delay: Duration): Flow<Unit> = channelFlow {
    delay(delay)

    if (isActive) send(Unit)
}

/**
 * Polls the [Flow] retrieved via the provided [flowGetter] function, by invoking the [flowGetter]
 * function after the provided [initialDelay] and then consistently after the provided [period],
 * and using the provided polling [strategy].
 */
@FlowPreview
@ExperimentalCoroutinesApi
@ExperimentalTime
fun <T> F.poll(
    initialDelay: Duration = 0.seconds,
    period: Duration,
    strategy: FlatMapStrategy = FlatMapStrategy.Latest,
    flowGetter: suspend (Long) -> Flow<T>
): Flow<T> {
    val intervalFlow = F.timeInterval(initialDelay = initialDelay, period = period)

    return intervalFlow.flatMap(strategy = strategy) { flowGetter(it) }
}



/**
 * Flow that emits [A] every [period] while taking into account how much time it takes downstream to consume the emission.
 * If downstream takes longer to process than [period] than it immediately emits another [A].
 *
 * Use `onEach { delay(timeMillis) }` for an alternative that sleeps [period] between every element.
 * This is different in that the time between every element is equal to the specified period,
 * regardless of how much time it takes to process that tick downstream.
 *
 * i.e, for a period of 1 second and a delay(100), the timestamps of the emission would be 1s, 2s, 3s, ... when using [fixedRate].
 * Whereas with `onEach { delay(timeMillis) }` it would run at timestamps 1s, 2.1s, 3.2s, ...
 *
 * @param period period between [Unit] emits of the resulting [Flow].
 */
@ExperimentalTime
 fun <A> Flow<A>.metered(period: Duration): Flow<A> =
    fixedRate(period).zip(this) { _, a -> a }

@ExperimentalTime
 fun <A> Flow<A>.metered(period: Long): Flow<A> =
    fixedRate(period).zip(this) { _, a -> a }

@ExperimentalTime
 fun fixedRate(
    periodInMillis: Long,
    dampen: Boolean = true,
    timeStamp: () -> ComparableTimeMark = { TimeSource.Monotonic.markNow() }
): Flow<Unit> =
    fixedRate(periodInMillis.milliseconds, dampen, timeStamp)

/**
 * Flow that emits [Unit] every [period] while taking into account how much time it takes downstream to consume the emission.
 * If downstream takes longer to process than [period] than it immediately emits another [Unit],
 * if you set [dampen] to false it will send `n = downstreamTime / period` [Unit] elements immediately.
 *
 * Use `onEach { delay(timeMillis) }` for an alternative that sleeps [period] between every element.
 * This is different in that the time between every element is equal to the specified period,
 * regardless of how much time it takes to process that tick downstream.
 *
 * i.e, for a period of 1 second and a delay(100), the timestamps of the emission would be 1s, 2s, 3s, ... when using [fixedRate].
 * Whereas with `onEach { delay(timeMillis) }` it would run at timestamps 1s, 2.1s, 3.2s, ...
 *
 * @param period period between [Unit] emits of the resulting [Flow].
 * @param dampen if you set [dampen] to false it will send `n` times [period] time it took downstream to process the emission.
 * @param timeStamp allows for supplying a different timestamp function, useful to override with `runBlockingTest`
 */
 fun fixedRate(
    period: Duration,
    dampen: Boolean = true,
    timeStamp: () -> ComparableTimeMark = { TimeSource.Monotonic.markNow() }
): Flow<Unit> =
    if (period == Duration.ZERO) flowOf(Unit).repeat()
    else flow {
        var lastAwakeAt = timeStamp()

        while (true) {
            val now = timeStamp()
            val next = lastAwakeAt + period

            if (next > now) {
                delay(next - now)
                emit(Unit)
                lastAwakeAt = next
            } else {
                val ticks: Long = ((now - lastAwakeAt).inWholeMilliseconds - 1) / period.inWholeMilliseconds
                when {
                    ticks < 0L -> Unit
                    ticks == 0L || dampen -> emit(Unit)
                    else -> repeat(ticks.toInt()) { emit(Unit) }
                }
                lastAwakeAt += (period * ticks.toDouble())
            }
        }
    }


/** Emits values from a [factory], with a [delay] between each emission. */
@ExperimentalTime
internal inline fun <T> ticker(
    delay: Duration,
    crossinline factory: () -> T,
): Flow<T> =
    flow {
        val marker = TimeSource.Monotonic
        while (true) {
            val before = marker.markNow()
            emit(factory.invoke())
            val adjustedDelay = delay - before.elapsedNow()
            if (adjustedDelay > delay) {
                // Something strange happened and the monotonic clock failed. In
                // such a case, don't delay and re-emit immediately. Usually this
                // will be a one-time thing like a timezone change.
            } else {
                // Negative delays are ignored, so no need to check if the clock
                // did something weird in the other direction.
                delay(adjustedDelay)
            }
        }
    }.conflate()

